package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.hmdp.constant.RedisConstants;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.SnowFlakeUtil;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Service
@Slf4j
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private SnowFlakeUtil snowFlakeUtil;
    @Autowired
    private ISeckillVoucherService seckillVoucherService;
    private static  final DefaultRedisScript<Long> SECKILL_SCRIPT ;
    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setResultType(Long.class);
        SECKILL_SCRIPT.setLocation(new ClassPathResource("lua/seckill.lua"));
    }

    private static final ExecutorService executor = Executors.newSingleThreadExecutor();

    @PostConstruct
    public void init() {
        executor.execute(new VoucherOrderHandler());
    }

    /**
     * 秒杀优惠券
     * @param voucherId
     * @return
     */
    @Override
    public Long seckillVoucher(Long voucherId) {
        log.info("开始秒杀");
        //雪花算法生成全局唯一id
        Long orderId = snowFlakeUtil.nextId();
        //执行lua脚本，进行资格判断
        Long userId= UserHolder.getUser().getId();
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.EMPTY_LIST,
                voucherId.toString(),
                userId.toString(),
                orderId.toString());

        if(result != 0L){
            if(result==1L) throw new RuntimeException("库存不足，秒杀失败");
            else throw new RuntimeException("重复下单，秒杀失败");
        }

        return orderId;
    }

    /**
     * 子线程进行写操作
     */
    private class VoucherOrderHandler implements Runnable {
        @Override
        public void run() {
            //死循环持续监听
            while (true) {
                //从消息队列取出消息
                List<MapRecord<String, Object, Object>> queue = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                        StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
                );
                //队列为空，进行下一次循环
                if(queue.isEmpty()||queue.size()==0){
                    continue;
                }
                //不为空，解析消息
                MapRecord<String, Object, Object> record = queue.get(0);
                Map<Object, Object> value = record.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), false);
                try {
                    //处理消息
                    createVoucherOrder(voucherOrder);
                } catch (Exception e) {
                    //处理失败
                    log.error("生成订单异常");
                    handlePendingList();
                }
                //处理成功，确认消息
                stringRedisTemplate.opsForStream().acknowledge("stream.orders","g1",record.getId());
            }
    }

        /**
         * 生成订单
         * @param voucherOrder
         */
        @Transactional
    public void createVoucherOrder(VoucherOrder voucherOrder) {
            try {
                //扣减库存
                boolean success = seckillVoucherService.update().setSql("stock = stock-1")
                        .eq("voucher_id", voucherOrder.getVoucherId())
                        .gt("stock", 0)
                        .update();
                if(!success){
                    log.error("扣减库存失败");
                }
                //生成订单
                save(voucherOrder);
            } catch (Exception e) {
                stringRedisTemplate.opsForValue().increment(RedisConstants.SECKILL_STOCK_KEY+voucherOrder.getVoucherId(),1);
                stringRedisTemplate.opsForSet().remove("seckill:order:"+voucherOrder.getVoucherId(),voucherOrder.getUserId());
                throw new RuntimeException("创建订单失败");
            }

        }

        /**
         * 处理 pending list
         */
    public void handlePendingList(){
        while (true) {
            //从pendingList取出消息
            List<MapRecord<String, Object, Object>> queue = stringRedisTemplate.opsForStream().read(
                    Consumer.from("g1", "c1"),
                    StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                    StreamOffset.create("stream.orders", ReadOffset.from(String.valueOf(0)))
            );
            //队列为空，跳出循环
            if(queue.isEmpty()||queue.size()==0){
                break;
            }
            //不为空，解析消息
            MapRecord<String, Object, Object> record = queue.get(0);
            Map<Object, Object> value = record.getValue();
            VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), false);
            try {
                //处理消息
                createVoucherOrder(voucherOrder);
            } catch (Exception e) {
                //处理失败
                log.error("生成订单异常");
                try {
                    Thread.sleep(50);
                } catch (InterruptedException ex) {
                    throw new RuntimeException(ex);
                }
            }
            //处理成功，确认消息
            stringRedisTemplate.opsForStream().acknowledge("stream.orders","g1",record.getId());
        }
    }
    }
}
